// 2015年版权所有etcd作者
// 
// 根据Apache许可证2.0版（以下简称“许可证”）获得许可；
// 除非遵守许可证，否则不得使用此文件。
// 您可以通过
// 
// www.apache.org/licenses/LICENSE-2.0 
// 
// 除非适用法律要求或书面同意，否则根据许可证分发的软件
// 按“原样”分发，
// 无任何明示或暗示的保证或条件。
// 请参阅许可证，了解管理许可的特定语言和
// 许可证下的限制。

package backend

import (
	"bytes"
	"math"
	"sync"
	"sync/atomic"
	"time"

	bolt "go.etcd.io/bbolt"
	"go.uber.org/zap"
)

type BucketID int

//
type Bucket interface {
	// ID返回存储桶的唯一标识符。
	// 该id不能持久化，并且可以在内存映射中用作轻量级标识符
	// 。
	ID() BucketID
	Name() []byte
	// 字符串实现了Stringer（人类可读的名称）。
	String() string

	// IsSafeRangeBucket是一种防止无意中读取重复密钥的黑客手段；
	// 在一个存储桶上覆盖的内容只能在limit=1的情况下获取，但已知safeRangeBucket 
	// 不会覆盖任何密钥，因此范围是安全的。
	IsSafeRangeBucket() bool
}

// etcd对批量读写事务的抽象
type BatchTx interface {
	ReadTx	// 内嵌了ReadTx接口
	UnsafeCreateBucket(bucket Bucket)	// 创建Bucket
	UnsafeDeleteBucket(bucket Bucket)
	// 向指定Bucket中添加键值对，与UnsafePut（）方法的区别是，
	// 其中会将对应Bucket实例的填充比例设置为90%，这样可以在顺序写入时，提高Bucket的利用率
	UnsafePut(bucket Bucket, key []byte, value []byte)
	UnsafeSeqPut(bucket Bucket, key []byte, value []byte)
	// 在指定Bucket中删除指定的键位对
	UnsafeDelete(bucket Bucket, key []byte)
	// 提交当前的读写事务，之后立即打开一个新的读写事务
	Commit()
	// 提交当前的读写事务，之后并不会再打开新的读写事务
	CommitAndStop()
}

// 在batchTx结构体中内嵌了sync.Mut饵，并且在batchTx和batchTxBuffered中都重写了其Unlock（）方法。
type batchTx struct {
	sync.Mutex
	// 该batchTx实例底层封装的bolt.Ix实例，即BoltDB层面的读写事务。
	tx      *bolt.Tx
	// 该batchTx实例关联的backend实例。
	backend *backend
	// 当前事务中执行的修改操作个数，在当前读写事务提交时，该值会被重置为0。
	pending int
}

func (t *batchTx) Lock() {
	t.Mutex.Lock()
}

func (t *batchTx) Unlock() {
	if t.pending >= t.backend.batchLimit {	// 检测当前事务的修改操作数量是否达到上限
		t.commit(false)	// 提交当前读写事务，并开启新的事务
	}
	t.Mutex.Unlock()	// 释放锁
}

// BatchTx接口嵌入ReadTx接口。但是RLock（）和RUnlock（）在BatchTx接口中没有合适的语义。因此不应被称为。
// TODO:可能希望将ReadTx和BatchTx解耦

func (t *batchTx) RLock() {
	panic("unexpected RLock")
}

func (t *batchTx) RUnlock() {
	panic("unexpected RUnlock")
}

// 是直接调用BoltDB的API创建了相应的Bucket实例
func (t *batchTx) UnsafeCreateBucket(bucket Bucket) {
	_, err := t.tx.CreateBucket(bucket.Name())
	if err != nil && err != bolt.ErrBucketExists {
		t.backend.lg.Fatal(
			"failed to create a bucket",
			zap.Stringer("bucket-name", bucket),
			zap.Error(err),
		)
	}
	t.pending++
}

func (t *batchTx) UnsafeDeleteBucket(bucket Bucket) {
	err := t.tx.DeleteBucket(bucket.Name())
	if err != nil && err != bolt.ErrBucketNotFound {
		t.backend.lg.Fatal(
			"failed to delete a bucket",
			zap.Stringer("bucket-name", bucket),
			zap.Error(err),
		)
	}
	t.pending++
}

// UnsafePut必须被称为保持tx上的锁。
func (t *batchTx) UnsafePut(bucket Bucket, key []byte, value []byte) {
	t.unsafePut(bucket, key, value, false)
}

// UnsafeSeqPut必须被称为保持tx上的锁。
func (t *batchTx) UnsafeSeqPut(bucket Bucket, key []byte, value []byte) {
	t.unsafePut(bucket, key, value, true)
}

func (t *batchTx) unsafePut(bucketType Bucket, key []byte, value []byte, seq bool) {
	bucket := t.tx.Bucket(bucketType.Name())	// 通过BoltDB提供的API获取指定的Bucket实例
	if bucket == nil {
		t.backend.lg.Fatal(
			"failed to find a bucket",
			zap.Stringer("bucket-name", bucketType),
			zap.Stack("stack"),
		)
	}
	if seq {	// 如果是顺序写入,则将填充率设置成90%
		bucket.FillPercent = 0.9
	}
	// 调用BoltDB提供的API写入键值对，
	if err := bucket.Put(key, value); err != nil {
		t.backend.lg.Fatal(
			"failed to write to a bucket",
			zap.Stringer("bucket-name", bucketType),
			zap.Error(err),
		)
	}
	t.pending++
}

// 必须调用UnsafeRange以保持发送上的锁。
func (t *batchTx) UnsafeRange(bucketType Bucket, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
	bucket := t.tx.Bucket(bucketType.Name())
	if bucket == nil {
		t.backend.lg.Fatal(
			"failed to find a bucket",
			zap.Stringer("bucket-name", bucketType),
			zap.Stack("stack"),
		)
	}
	return unsafeRange(bucket.Cursor(), key, endKey, limit)
}

func unsafeRange(c *bolt.Cursor, key, endKey []byte, limit int64) (keys [][]byte, vs [][]byte) {
	if limit <= 0 {
		limit = math.MaxInt64
	}
	var isMatch func(b []byte) bool
	if len(endKey) > 0 {
		isMatch = func(b []byte) bool { return bytes.Compare(b, endKey) < 0 }
	} else {
		isMatch = func(b []byte) bool { return bytes.Equal(b, key) }
		limit = 1
	}

	for ck, cv := c.Seek(key); ck != nil && isMatch(ck); ck, cv = c.Next() {
		vs = append(vs, cv)	// 记录符合条件的value值
		keys = append(keys, ck)	// 记录符合条件的key佳
		if limit == int64(len(keys)) {
			break
		}
	}
	return keys, vs
}

// 通过BoltDBAPI查找到对应的Bucket实例，然后通过其Delete（）方法删除键值对
func (t *batchTx) UnsafeDelete(bucketType Bucket, key []byte) {
	bucket := t.tx.Bucket(bucketType.Name())
	if bucket == nil {
		t.backend.lg.Fatal(
			"failed to find a bucket",
			zap.Stringer("bucket-name", bucketType),
			zap.Stack("stack"),
		)
	}
	err := bucket.Delete(key)
	if err != nil {
		t.backend.lg.Fatal(
			"failed to delete a key",
			zap.Stringer("bucket-name", bucketType),
			zap.Error(err),
		)
	}
	t.pending++
}


func (t *batchTx) UnsafeForEach(bucket Bucket, visitor func(k, v []byte) error) error {
	return unsafeForEach(t.tx, bucket, visitor)
}

// 调用Bucket.ForEach（）方法实现对BoltDB中键值对的遍历，
func unsafeForEach(tx *bolt.Tx, bucket Bucket, visitor func(k, v []byte) error) error {
	if b := tx.Bucket(bucket.Name()); b != nil {	// 查找指定的Bucket实例
		return b.ForEach(visitor)	// 调用Bucket.ForEach（）方法进行遍历
	}
	return nil
}

// 处理读写事务
func (t *batchTx) Commit() {
	t.Lock()
	t.commit(false)
	t.Unlock()
}

// CommitAndStop提交以前的发送，不创建新的发送。
func (t *batchTx) CommitAndStop() {
	t.Lock()
	t.commit(true)
	t.Unlock()
}

func (t *batchTx) safePending() int {
	t.Mutex.Lock()
	defer t.Mutex.Unlock()
	return t.pending
}

// 处理读写事务
func (t *batchTx) commit(stop bool) {
	if t.tx != nil {
		if t.pending == 0 && !stop {	// 当前读写事务中未进行任何修改操作，则无须开启新事务
			return
		}

		start := time.Now()

		// gofail:var beforeCommit struct{}
		err := t.tx.Commit()	// 通过BoltDB提供的API提交当前读写事务
		// gofail:var after commit struct{}

		rebalanceSec.Observe(t.tx.Stats().RebalanceTime.Seconds())
		spillSec.Observe(t.tx.Stats().SpillTime.Seconds())
		writeSec.Observe(t.tx.Stats().WriteTime.Seconds())
		commitSec.Observe(time.Since(start).Seconds())
		atomic.AddInt64(&t.backend.commits, 1)	// 通过BoltDB提供的API提交当前读写事务

		t.pending = 0	// 重置pending字段
		if err != nil {
			t.backend.lg.Fatal("failed to commit tx", zap.Error(err))
		}
	}
	if !stop {
		t.tx = t.backend.begin(true)	// 开启新的读写事务
	}
}

// batchTxBuffered中内嵌了batchTx结构体，同时还包含一个buf字段（txWriteBuffer）作为缓存，
// 在backend使用的是batchTxBuffered实例，而不是直接使用batchTx实例
type batchTxBuffered struct {
	batchTx
	buf txWriteBuffer
}

// batchTxBuffered和batchTx的初始化
func newBatchTxBuffered(backend *backend) *batchTxBuffered {
	tx := &batchTxBuffered{
		batchTx: batchTx{backend: backend},	// 创建内嵌的batchTx实例
		buf: txWriteBuffer{	// 创建txWriteBuffer缓冲区
			txBuffer:   txBuffer{make(map[BucketID]*bucketBuffer)},
			bucket2seq: make(map[BucketID]bool),
		},
	}
	tx.Commit()	// 开启一个读写事务，
	return tx
}

// batchTxBuffered.Unlock（）方法中，除了调用batchTx.Unlock（）方法实现解锁，
// 还会将当前batchTxBuffered中缓存的键值对更新到readTx缓存中
// （即通过tx WriteBuffer. writeback（）方法），从而实现只读事务的缓存更新。
// 这也是UnsafePut*（）等方法在执行前必须获取锁的原因。
func (t *batchTxBuffered) Unlock() {
	if t.pending != 0 {	// 检测当前读写事务中是否发生了修改操作
		t.backend.readTx.Lock() // 更新readTx的缓存
		t.buf.writeback(&t.backend.readTx.buf)
		t.backend.readTx.Unlock()
		// 如果当前事务的修改操作数量是否达到上限，则提交当前事务，开启新事务
		if t.pending >= t.backend.batchLimit {
			t.commit(false)
		}
	}
	// 调用batchTx.Unlock（）方法完成解锁
	t.batchTx.Unlock()
}

func (t *batchTxBuffered) Commit() {
	t.Lock()
	t.commit(false)
	t.Unlock()
}

func (t *batchTxBuffered) CommitAndStop() {
	t.Lock()
	t.commit(true)
	t.Unlock()
}

func (t *batchTxBuffered) commit(stop bool) {
	if t.backend.hooks != nil {
		t.backend.hooks.OnPreCommitUnsafe(t)
	}

	// 必须关闭所有读取tx才能获取boltdb commit rwlock 
	t.backend.readTx.Lock()
	t.unsafeCommit(stop)
	t.backend.readTx.Unlock()
}

// 先回滚当前的只读事务，提交当前的读写事务，然后开启新的只读事务和读写事务。
// 注意，在Commit（）方法中需要更新readTx和batchTxBuffered的宇段（例如，tx字段、buf字段及其他的统计字段等），
// 所以需要先获取readTx和batchTxBuffered的锁进行同步。
func (t *batchTxBuffered) unsafeCommit(stop bool) {
	if t.backend.readTx.tx != nil {
		// 如采当前已经开启了只读事务，则将该事务回滚（BoltDB中的只读事务只能回滚，无法提交）
		go func(tx *bolt.Tx, wg *sync.WaitGroup) {
			wg.Wait()
			if err := tx.Rollback(); err != nil {
				t.backend.lg.Fatal("failed to rollback tx", zap.Error(err))
			}
		}(t.backend.readTx.tx, t.backend.readTx.txWg)
		t.backend.readTx.reset()	// 清空readTx中的缓存
	}
	// 如果当前已经开启了读写事务，则将该事务提交，并创建新的读写事务
	t.batchTx.commit(stop)
	// 根据stop参数，决定事务开启新的只读事务
	if !stop {
		t.backend.readTx.tx = t.backend.begin(false)
	}
}

// 调用batchTx中对应的方法将键值对写入BoltDB，还会调用其buf宇段（txWriteBuffer类型）的方法，
// 将键值对写入batchTxBuffered的缓存中
func (t *batchTxBuffered) UnsafePut(bucket Bucket, key []byte, value []byte) {
	t.batchTx.UnsafePut(bucket, key, value)	// 将键值对写入BoltDB
	t.buf.put(bucket, key, value)	// 将键值对写入缓存
}

func (t *batchTxBuffered) UnsafeSeqPut(bucket Bucket, key []byte, value []byte) {
	t.batchTx.UnsafeSeqPut(bucket, key, value)	// 将键值对写入BoltDB
	t.buf.putSeq(bucket, key, value)	// 将键值对写入缓存
}
